package com.niodata.dt.fs;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;

/**
 * relative root dir local filesystem.
 * file operation path is relative to root dir.
 */
public class DpRelativeLocalFileSystem extends FileSystem {

  public static final String FS_DP_REL_LOCAL_ROOTDIR = "fs.rel.local.root";
  private URI uri;
  private Path workingDir;
  private String rootPath;


  @Override
  public void initialize(URI name, Configuration conf) throws IOException {
    super.setConf(conf);
    super.initialize(name, conf);
    this.rootPath = conf.get(FS_DP_REL_LOCAL_ROOTDIR);
    if (rootPath.endsWith("/")) {
      rootPath = rootPath.substring(0, rootPath.length() - 1);
    }
    File file = new File(this.rootPath);
    if (!file.exists()) {
      throw new FileNotFoundException(file.getAbsolutePath());
    }
    if (file.isFile()) {
      throw new IOException("root path is file:" + file.getAbsolutePath());
    }
    String url = "dpRelLocal://" + this.rootPath;
    this.uri = URI.create(url);
    workingDir = new Path(System.getProperty("user.dir"));
  }


  @Override
  public URI getUri() {
    return this.uri;
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    f = this.buildFullPath(f);
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    File file = new File(path);
    return new FSDataInputStream(new LocalFsInputStream(file, new FileInputStream(file)));
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize,
                                   short replication, long blockSize, Progressable progress) throws IOException {
    f = buildFullPath(f);
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    return new FSDataOutputStream(new FileOutputStream(path), null);
  }

  @Override
  public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException {
    f = this.buildFullPath(f);
    String path = Path.getPathWithoutSchemeAndAuthority(f).toString();
    return new FSDataOutputStream(new FileOutputStream(path, true), null);
  }

  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    src = this.buildFullPath(src);
    dst = this.buildFullPath(dst);
    File source = new File(Path.getPathWithoutSchemeAndAuthority(src).toString());
    File target = new File(Path.getPathWithoutSchemeAndAuthority(dst).toString());
    return source.renameTo(target);
  }

  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    f = this.buildFullPath(f);
    File file = new File(f.toString());
    FileUtil.deleteFileOrDir(file, recursive);
    return true;
  }

  @Override
  public FileStatus[] listStatus(Path f) throws IOException {
    f = buildFullPath(f);
    File file = new File(f.toString());
    if (file.isFile()) {
      throw new IOException("path is file:" + f);
    }
    File[] files = file.listFiles();
    if (files == null) {
      return new FileStatus[0];
    }
    FileStatus[] statuses = new FileStatus[files.length];
    for (int i = 0; i < files.length; i++) {
      File tf = files[i];
      FileStatus status = null;
      String relPath = tf.getAbsolutePath().substring(rootPath.length());
      Path path = new Path(relPath);
      if (tf.isFile()) {
        status = new FileStatus(tf.length(), false, 1,
          FileUtil.SIZE_128MB, tf.lastModified(), path);
      } else {
        status = new FileStatus(0, true, 1,
          FileUtil.SIZE_128MB, tf.lastModified(), path);
      }
      statuses[i] = status;
    }
    return statuses;
  }

  @Override
  public Path getWorkingDirectory() {
    return workingDir;
  }

  @Override
  public void setWorkingDirectory(Path newdir) {
    this.workingDir = newdir;
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    f = buildFullPath(f);
    File file = new File(f.toString());
    return file.mkdirs();
  }

  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    Path fullPath = buildFullPath(f);
    File tf = new File(fullPath.toString());
    if (!tf.exists()) {
      throw new FileNotFoundException(f.toString());
    }
    FileStatus status;
    if (tf.isFile()) {
      status = new FileStatus(tf.length(), false, 1,
        FileUtil.SIZE_128MB, tf.lastModified(), f);
    } else {
      status = new FileStatus(0, true, 1,
        FileUtil.SIZE_128MB, tf.lastModified(), f);
    }
    return status;
  }

  private Path buildFullPath(Path relPath) {
    String path = Path.getPathWithoutSchemeAndAuthority(relPath).toString();
    String fullPath = this.rootPath + path;
    return new Path(fullPath);
  }

  @Override
  public void copyToLocalFile(Path src, Path dst) throws IOException {
    super.copyToLocalFile(false, src, dst, true);
  }

  @Override
  public void copyToLocalFile(boolean delSrc, Path src, Path dst) throws IOException {
    super.copyToLocalFile(delSrc, src, dst, true);
  }
}
